package com.nebula.stream;

import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.messaging.handler.annotation.SendTo;

/**
 * @Author: llq
 * @Date: 2019/9/26
 *
 * @EnableBinding(Sink.class) 开启了消息通道实现的（Sink/soutce/Processor，可以自定义）的binding(绑定),连接到了消息代理
 * 这样做会向框架发出信号，以启动与消息传递中间件的绑定，并自动创建绑定到通道的目标（即queue，topic，exchange和其他）
 * 目的地绑定，提供消息中间件与生产者与消费者的连接
 */
@Slf4j
@SpringBootApplication
@EnableBinding({Processor.class,MyPipe.class})
public class NeStream {
    public static void main(String[] args) {
        SpringApplication.run(NeStream.class,args);
    }

    /**
     * @StreamListener 监听消息类型为String的消息，这么做是为了展示框架的核心特性之一——自动转换入参消息体为指定类型
     * 由  通道接收消息转换成String 类型并打印在控制台上，然后转换出一个大写的信息返回到OUTPUT中
     * @SendTo 将方法内返回值转发到其他消息通道中
     * @param value
     */
    @StreamListener(Processor.INPUT)
    @SendTo(MyPipe.INPUT)
    public String receive(String value) {
        log.info("Received: " + value);
        System.out.println("Received: " + value);
        return value.toUpperCase();
    }


    @StreamListener(MyPipe.INPUT)
    public void receiveFromMyPipe(String value){
        log.info("Received: "+value);
        System.out.println("Received: " + value);
    }
}
